package com.woniu.intlet.timer;

import com.woniu.config.RabbitMqTemplate;
import com.woniu.outlet.dao.MessagePoMapper;
import com.woniu.outlet.po.MessagePo;
import com.woniu.outlet.po.MessagePoExample;
import com.woniu.uitl.MessageStatus;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * 基于实现rabbimtmq解决分布式事务最终一致性的定时器
 */
@Component
@Slf4j
public class MessageTimer {

    @Autowired
    private MessagePoMapper messagePoMapper;
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Scheduled(fixedRate = 10000)
    public void sendMessage(){
        //查询消息记录表中不是已经发送且次数未0的数据
        MessagePoExample messagePoExample = new MessagePoExample();
        MessagePoExample.Criteria criteria = messagePoExample.createCriteria();
        criteria.andStatusNotEqualTo(MessageStatus.SUCESS.getStatus());
        criteria.andRetryCountGreaterThan(0);
        List<MessagePo> messagePos = messagePoMapper.selectByExample(messagePoExample);

        //迭代重复发送消息
        for (MessagePo messagePo : messagePos) {
            MessagePostProcessor processor = new MessagePostProcessor() {
                @Override
                public Message postProcessMessage(Message message) throws AmqpException {
                    message.getMessageProperties().setClusterId(messagePo.getId().toString());
                    return message;
                }
            };
            rabbitTemplate.convertAndSend(messagePo.getExchange(),messagePo.getRoutingKey(),messagePo.getMessageContent(),processor,new CorrelationData(messagePo.getId().toString()));
            //发送次数-1
            messagePo.setRetryCount(messagePo.getRetryCount()-1);
            //更新到数据库
            messagePoMapper.updateByPrimaryKeySelective(messagePo);
        }
    }

}
